using CSV, DataFrames, Dagger, Statistics

## prepare data

dir = mktempdir()

mean_squared_input = Float64[]
for sample_idx in 1:5
   x = rand(10)
   CSV.write("$(dir)/sample_$(sample_idx).csv", DataFrame(x=x))
   push!(mean_squared_input, mean(x.^2))
end

samples = ["$(dir)/sample_$(sample_idx).csv" for sample_idx in 1:5]

## define rules 

# function that creates a Rule for a given sample
get_rule_square(sample) = Dagger.Rule(sample => replace(sample, "sample_" => "sample_squared_"); forcerun=false) do input, output
   df = CSV.read(input[1], DataFrame)
   df.xsquared = df.x .^ 2
   CSV.write(output[1], df)
   output
end

squared_rules = get_rule_square.(samples)
squared_rule_outputs = [only(r.outputs) for r in squared_rules]

make_summary = Dagger.Rule(squared_rule_outputs => "$(dir)/samples_summary.csv"; forcerun=false) do inputs, output
   dfs = CSV.read.(inputs, DataFrame)
   mean_squared = DataFrame(sample = inputs, mean_squared = [mean(df.xsquared) for df in dfs])
   CSV.write(output[1], mean_squared)
   output
end

## Run 

squared = [Dagger.@spawn r() for r in squared_rules]
summary_file = Dagger.@spawn make_summary(squared...)

out = CSV.read(fetch(summary_file), DataFrame)

@assert out.mean_squared == mean_squared_input

@assert Dagger.needs_update(make_summary) == false
run(`touch $(squared_rule_outputs[1])`)
sleep(0.5)
@assert Dagger.needs_update(make_summary) == true

run(`rm $(squared_rule_outputs[1])`)
summary_file = Dagger.@spawn make_summary(squared...) # fails

squared = [Dagger.@spawn r() for r in squared_rules] # redo only 1 file

summary_file = Dagger.@spawn make_summary(squared...)